Workflowsで使えるBigQueryのqueryコネクタを用いてBigQueryにクエリ発行する時はtimeoutMsフィールドに注意

Method: googleapis.bigquery.v2.jobs.query

Runs a BigQuery SQL query synchronously and returns query results if the query completes within a specified timeout.
BigQueryのSQLクエリを同期的に実行し、指定したタイムアウト内にクエリが完了した場合にクエリ結果を返します。

import json
import random
import string
import os

# 3GBの目標サイズ（バイト単位）
TARGET_SIZE = 3 * 1024 * 1024 * 1024

# 出力ファイル名
OUTPUT_FILE = "large_file.jsonl"

def random_string(length=10):
    """ランダムな文字列を生成する"""
    return ''.join(random.choices(string.ascii_letters + string.digits, k=length))

def random_record():
    """ランダムなレコードを生成する"""
    return {
        "id": random.randint(1, 1000000),
        "name": random_string(10),
        "email": f"{random_string(5)}@example.com",
        "age": random.randint(18, 99),
        "address": {
            "city": random_string(8),
            "zipcode": random.randint(10000, 99999),
        },
        "is_active": random.choice([True, False]),
        "tags": [random_string(5) for _ in range(random.randint(1, 5))]
    }

def create_large_jsonlines_file():
    """JSONLinesファイルを作成する"""
    with open(OUTPUT_FILE, "w") as f:
        current_size = 0

        while current_size < TARGET_SIZE:
            record = random_record()
            json_record = json.dumps(record)
            f.write(json_record + "\n")  # 各レコードを1行に書き込む

            current_size += len(json_record.encode("utf-8")) + 1  # +1は改行文字の分

    print(f"JSONLinesファイル '{OUTPUT_FILE}' が作成されました。サイズ: {os.path.getsize(OUTPUT_FILE) / (1024 * 1024):.2f} MB")

if __name__ == "__main__":
    create_large_jsonlines_file()

CREATE OR REPLACE EXTERNAL TABLE tmp_nemoto.test(
  id INT64,
  name STRING,
  email STRING,
  age INT64,
  address STRUCT<
    city STRING,
    zipcode INT64
  >,
  is_active BOOL,
  tags ARRAY<STRING>
)
  OPTIONS (
  format = 'json',
  uris = ['gs://バケット名/bq_data/large_file*.jsonl']);

SELECT
  COUNT(*) AS table_count
FROM データセット名.test

- init:
    assign:
        - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
        - sqlBucket: "バケット名"
        - sqlFilePath: "SQLファイル名"
- readSqlFiles:
    call: googleapis.storage.v1.objects.get
    args:
        bucket: ${sqlBucket}
        object: ${text.url_encode(sqlFilePath)}
        alt: "media"
    result: sqlFileContent
- executeQuery:
    call: googleapis.bigquery.v2.jobs.query
    args:
        projectId: ${project_id}
        body:
            query: ${text.decode(sqlFileContent)}
            useLegacySql: false
            useQueryCache: false
            #timeoutMs: 20000 まずはコメントアウト
    result: queryResult
- executeQueryLog:
    call: sys.log
    args:
        text: ${queryResult}
        severity: "INFO"
- returnStep:
    return: ${queryResult}

{
  "jobComplete": false,
  "jobCreationReason": {
    "code": "REQUESTED"
  },
  "jobReference": {
    "jobId": "job_8vPjkjEh-3lzGFq0gW6hGl3RwQW7",
    "location": "asia-northeast1",
    "projectId": "プロジェクトID"
  },
  "kind": "bigquery#queryResponse",
  "queryId": "job_8vPjkjEh-3lzGFq0gW6hGl3RwQW7"
}

SELECT
  job_id,
  state,
  TIMESTAMP_DIFF(end_time, start_time, SECOND) AS execution_time_seconds
FROM
  `region-asia-northeast1`.`INFORMATION_SCHEMA.JOBS`
WHERE job_id = 'job_8vPjkjEh-3lzGFq0gW6hGl3RwQW7';

- init:
    assign:
        - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
        - sqlBucket: "バケット名"
        - sqlFilePath: "SQLファイル名"
- readSqlFiles:
    call: googleapis.storage.v1.objects.get
    args:
        bucket: ${sqlBucket}
        object: ${text.url_encode(sqlFilePath)}
        alt: "media"
    result: sqlFileContent
- executeQuery:
    call: googleapis.bigquery.v2.jobs.query
    args:
        projectId: ${project_id}
        body:
            query: ${text.decode(sqlFileContent)}
            useLegacySql: false
            useQueryCache: false
            timeoutMs: 20000 #コメントアウト解除
    result: queryResult
- executeQueryLog:
    call: sys.log
    args:
        text: ${queryResult}
        severity: "INFO"
- returnStep:
    return: ${queryResult}

{
  "cacheHit": false,
  "jobComplete": true,
  "jobCreationReason": {
    "code": "REQUESTED"
  },
  "jobReference": {
    "jobId": "job_kXqnrVt-7mxMmB0sde_ausdLsbnJ",
    "location": "asia-northeast1",
    "projectId": "プロジェクトID"
  },
  "kind": "bigquery#queryResponse",
  "queryId": "job_kXqnrVt-7mxMmB0sde_ausdLsbnJ",
  "rows": [
    {
      "f": [
        {
          "v": "17376189"
        }
      ]
    }
  ],
  "schema": {
    "fields": [
      {
        "mode": "NULLABLE",
        "name": "table_count",
        "type": "INTEGER"
      }
    ]
  },
  "totalBytesProcessed": "3221225601",
  "totalRows": "1"
}